package cn.gwm.flink.streaming.strategy.vehiclepredict;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * @author : GW00280745
 * @date : 2022/8/18 13:05
 * @description :
 */
public class ChargePredictProducer extends RichSinkFunction {
    private Properties prodProps;
    private String topic;
    public ChargePredictProducer(Properties prodProps, String topic){
        this.prodProps = prodProps;
        this.topic = topic;
        //2M 或 1000ms 进行一次 producer.flush() 负载高用2M负载低用ms
        prodProps.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"23840");
        prodProps.setProperty(ProducerConfig.LINGER_MS_CONFIG,"1000");
        //消息发送失败后重试次数=5
        prodProps.setProperty(ProducerConfig.RETRIES_CONFIG,"5");
        //Step1：异步调用send发送日志，根据Properties的配置对kv进行序列化
        //Step2：根据k hash 得到分区信息，追加到对应topic下的partition，这里先会写入到本地缓存区
        //Step3：本地缓存写入后，有独立的线程传送向producer发送ACK
    }
    KafkaProducer<String, String> producer;
    @Override
    public void open(Configuration parameters) throws Exception {
        producer = new KafkaProducer<String, String>(prodProps);
    }

    @Override
    public void close() throws Exception {
        if (producer==null){
            return;
        }
        producer.flush();
        producer.close();
    }
    private String bodyStr = "body";
    private String vinStr = "vin";
    private String deviceIdStr = "deviceId";
    @Override
    public void invoke(Object value, Context context) throws Exception {
        try {
            JSONObject line = null;
            if (value instanceof String){
                line =JSONObject.parseObject(value.toString());
            }else if (value instanceof JSONObject){
                line =(JSONObject) value;
            }
            if (line==null){
                return;
            }
            String vin =null;
            if (line.containsKey(bodyStr)){
                JSONObject source = line.getJSONObject(bodyStr).getJSONObject("source");
                if (source.containsKey(vinStr)){
                    vin = source.getString(vinStr);
                }else if (source.containsKey(deviceIdStr)){
                    vin = source.getString(deviceIdStr);
                }
            }else if (line.containsKey(deviceIdStr)){
                vin = line.getString(deviceIdStr);
            }
            if (StringUtils.isBlank(vin)){
                return;
            }
            ProducerRecord<String, String> record = new ProducerRecord<>(topic,vin, line.toString());
            Future<RecordMetadata> future = producer.send(record);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}
